package com.norma.netty.study.netty.disruptor.core;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import com.norma.netty.study.netty.disruptor.TransDataWrapper;

import java.time.LocalDateTime;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author zhangchaofeng
 * @date 2021/1/5
 * @description 多生产多消费模型
 */
public class RingBufferWorkerPoolFactory {

    private ExecutorService eventHandlerPool = new ThreadPoolExecutor(5, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(20),
            new RejectPolicy(1000));

    private static class SingletonHolder {
        static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
    }

    private RingBufferWorkerPoolFactory() {
        System.out.println("时间=" + LocalDateTime.now());
    }

    public static RingBufferWorkerPoolFactory getInstance() {
        return SingletonHolder.instance;
    }

    private static Map<String, MessageConsumer> consumers = new ConcurrentHashMap<>();

    private static Map<String, MessageProducer> producers = new ConcurrentHashMap<>();

    private RingBuffer<TransDataWrapper> ringBuffer;

    private WorkerPool<TransDataWrapper> workerPool;

    private SequenceBarrier sequenceBarrier;

    public void initAndStart(ProducerType type, int bufferSize,
                             WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
        this.ringBuffer = RingBuffer.create(
                type,
                () -> new TransDataWrapper(),
                bufferSize,
                waitStrategy
        );
        this.sequenceBarrier = this.ringBuffer.newBarrier();
        this.workerPool = new WorkerPool(this.ringBuffer,
                this.sequenceBarrier, new EventExceptionHandler(), messageConsumers);

        for (MessageConsumer messageConsumer : messageConsumers) {
            consumers.put(messageConsumer.getConsumerId(), messageConsumer);
        }
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        this.workerPool.start(eventHandlerPool);
    }

    public MessageProducer getMessageProducer(String producerId) {
        MessageProducer messageProducer = producers.get(producerId);
        if (messageProducer == null) {
            messageProducer = new MessageProducer(producerId, ringBuffer);
            producers.put(producerId, messageProducer);
        }
        return messageProducer;
    }

    static class EventExceptionHandler implements ExceptionHandler<TransDataWrapper> {
        @Override
        public void handleEventException(Throwable throwable, long l, TransDataWrapper transDataWrapper) {

        }

        @Override
        public void handleOnStartException(Throwable throwable) {

        }

        @Override
        public void handleOnShutdownException(Throwable throwable) {

        }
    }

    public static class RejectPolicy implements RejectedExecutionHandler {
        private long waitTime;

        public RejectPolicy(long time) {
            this.waitTime = time;
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                Thread.sleep(waitTime);
            } catch (InterruptedException e) {
                // 丢弃异常
            }
            executor.execute(r);
        }
    }


}
